//
// Created by hope on 2020/12/22.
//

#include "db/db_test_util.h"
#include "options/options_helper.h"
#include "port/port.h"
#include "port/stack_trace.h"
#include "util/fault_injection_test_env.h"
#include "util/sync_point.h"

namespace rocksdb {
class DBPurememWALTest : public DBTestBase {
 public:
  DBPurememWALTest() : DBTestBase("/db_puremem_wal_test") {}

#if defined(ROCKSDB_PLATFORM_POSIX)
  uint64_t GetAllocatedFileSize(std::string file_name) {
    struct stat sbuf;
    int err = stat(file_name.c_str(), &sbuf);
    assert(err == 0);
    return sbuf.st_blocks * 512;
  }
#endif
 protected:
  Options GetPurememOptions(){
    auto db_options = CurrentOptions();
    db_options.pureMemTable = true;
    return db_options;
  }
};

// Jacob: 带有wal管理模块的db基本测试
TEST_F(DBPurememWALTest, WAL) {
  do {
    auto puremem_options = GetPurememOptions();
    CreateAndReopenWithCF({"pikachu"}, puremem_options);
    WriteOptions writeOpt = WriteOptions();
    writeOpt.disableWAL = false;
    ASSERT_OK(dbfull()->Put(writeOpt, handles_[1], "foo", "v1"));
    ASSERT_OK(dbfull()->Put(writeOpt, handles_[1], "bar", "v1"));
    ASSERT_EQ("v1", Get(1, "foo"));
    ASSERT_EQ("v1", Get(1, "bar"));
    ReopenWithColumnFamilies({"default", "pikachu"}, puremem_options);
    ASSERT_EQ("v1", Get(1, "foo"));
    ASSERT_EQ("v1", Get(1, "bar"));

    writeOpt.disableWAL = false;
    ASSERT_OK(dbfull()->Put(writeOpt, handles_[1], "bar", "v2"));
    writeOpt.disableWAL = true;
    ASSERT_OK(dbfull()->Put(writeOpt, handles_[1], "foo", "v2"));

    ReopenWithColumnFamilies({"default", "pikachu"}, puremem_options);

    ASSERT_EQ("v2", Get(1, "bar"));
    ASSERT_EQ("v1", Get(1, "foo"));

    writeOpt.disableWAL = true;
    ASSERT_OK(dbfull()->Put(writeOpt, handles_[1], "bar", "v3"));
    writeOpt.disableWAL = false;
    ASSERT_OK(dbfull()->Put(writeOpt, handles_[1], "foo", "v3"));

    ReopenWithColumnFamilies({"default", "pikachu"}, puremem_options);

    ASSERT_EQ("v3", Get(1, "foo"));
    ASSERT_EQ("v2", Get(1, "bar"));
    ASSERT_EQ("NOT_FOUND", Get(0, "foo"));
  } while (ChangeWalOptions());
}

// Jacob: 日志启动
TEST_F(DBPurememWALTest, RollLog) {
  do {
    auto puremem_options = GetPurememOptions();
    CreateAndReopenWithCF({"pikachu"}, puremem_options);
    ASSERT_OK(Put(1, "foo", "v1"));
    ASSERT_OK(Put(1, "baz", "v5"));

    ReopenWithColumnFamilies({"default", "pikachu"}, puremem_options);
    for (int i = 0; i < 10; i++) {
      ReopenWithColumnFamilies({"default", "pikachu"}, puremem_options);
    }
    ASSERT_OK(Put(1, "foo", "v4"));
    for (int i = 0; i < 10; i++) {
      ReopenWithColumnFamilies({"default", "pikachu"}, puremem_options);
    }
  } while (ChangeWalOptions());
}

// Jacob: WAL恢复
TEST_F(DBPurememWALTest, Recover) {
  do {
    auto puremem_options = GetPurememOptions();
    CreateAndReopenWithCF({"pikachu"}, puremem_options);
    ASSERT_OK(Put(1, "foo", "v1"));
    ASSERT_OK(Put(1, "baz", "v5"));

    ReopenWithColumnFamilies({"default", "pikachu"}, puremem_options);
    ASSERT_EQ("v1", Get(1, "foo"));

    ASSERT_EQ("v1", Get(1, "foo"));
    ASSERT_EQ("v5", Get(1, "baz"));
    ASSERT_OK(Put(1, "bar", "v2"));
    ASSERT_OK(Put(1, "foo", "v3"));

    ReopenWithColumnFamilies({"default", "pikachu"}, puremem_options);
    ASSERT_EQ("v3", Get(1, "foo"));
    ASSERT_OK(Put(1, "foo", "v4"));
    ASSERT_EQ("v4", Get(1, "foo"));
    ASSERT_EQ("v2", Get(1, "bar"));
    ASSERT_EQ("v5", Get(1, "baz"));
  } while (ChangeWalOptions());
}

// Jacob: 忽略日志文件，恢复不应该再次发生如果恢复发生，相同的合并操作符将被调用两次，导致错误的结果
TEST_F(DBPurememWALTest, IgnoreRecoveredLog) {
  std::string backup_logs = dbname_ + "/backup_logs";

  do {
    // delete old files in backup_logs directory
    env_->CreateDirIfMissing(backup_logs);
    std::vector<std::string> old_files;
    env_->GetChildren(backup_logs, &old_files);
    for (auto& file : old_files) {
      if (file != "." && file != "..") {
        env_->DeleteFile(backup_logs + "/" + file);
      }
    }
    auto puremem_options = GetPurememOptions();
    puremem_options.create_if_missing = true;
    puremem_options.merge_operator = MergeOperators::CreateUInt64AddOperator();
    puremem_options.wal_dir = dbname_ + "/logs";
    DestroyAndReopen(puremem_options);

    // fill up the DB
    ASSERT_OK(Put("foo", "v1"));
    ASSERT_OK(Put("baz", "v5"));
    ASSERT_EQ("v1", Get("foo"));
    ASSERT_EQ("v5", Get("baz"));

    // copy the logs to backup
    std::vector<std::string> logs;
    env_->GetChildren(puremem_options.wal_dir, &logs);
    for (auto& log : logs) {
      if (log != ".." && log != ".") {
        CopyFile(puremem_options.wal_dir + "/" + log, backup_logs + "/" + log);
      }
    }

    // recover the DB
    Reopen(puremem_options);
    ASSERT_EQ("v1", Get("foo"));
    ASSERT_EQ("v5", Get("baz"));
    Close();

    // copy the logs from backup back to wal dir
    for (auto& log : logs) {
      if (log != ".." && log != ".") {
        CopyFile(backup_logs + "/" + log, puremem_options.wal_dir + "/" + log);
      }
    }
    // this should ignore the log files, recovery should not happen again
    // if the recovery happens, the same merge operator would be called twice,
    // leading to incorrect results
    Reopen(puremem_options);
    ASSERT_EQ("v1", Get("foo"));
    ASSERT_EQ("v5", Get("baz"));
    Close();
    Destroy(puremem_options);
    Reopen(puremem_options);
    Close();

    // copy the logs from backup back to wal dir
    env_->CreateDirIfMissing(puremem_options.wal_dir);
    for (auto& log : logs) {
      if (log != ".." && log != ".") {
        CopyFile(backup_logs + "/" + log, puremem_options.wal_dir + "/" + log);
      }
    }
    // assert that we successfully recovered only from logs, even though we
    // destroyed the DB
    Reopen(puremem_options);
    ASSERT_EQ("v1", Get("foo"));
    ASSERT_EQ("v5", Get("baz"));

    Destroy(puremem_options);
    // copy the logs from backup back to wal dir
    env_->CreateDirIfMissing(puremem_options.wal_dir);
    for (auto& log : logs) {
      if (log != ".." && log != ".") {
        CopyFile(backup_logs + "/" + log, puremem_options.wal_dir + "/" + log);
        // we won't be needing this file no more
        env_->DeleteFile(backup_logs + "/" + log);
      }
    }
    Status s = TryReopen(puremem_options);
    ASSERT_EQ("v1", Get("foo"));
    ASSERT_EQ("v5", Get("baz"));
    Destroy(puremem_options);

    // Jacob: 如果没有WAL文件，将不会读入数据。
    Reopen(puremem_options);
    ASSERT_EQ("NOT_FOUND", Get("foo"));
    ASSERT_EQ("NOT_FOUND", Get("baz"));
  } while (ChangeWalOptions());
}

TEST_F(DBPurememWALTest, RecoveryWithEmptyLog) {
  do {
    auto puremem_options = GetPurememOptions();
    CreateAndReopenWithCF({"pikachu"}, puremem_options);
    ASSERT_OK(Put(1, "foo", "v1"));
    ASSERT_OK(Put(1, "foo", "v2"));
    ReopenWithColumnFamilies({"default", "pikachu"}, puremem_options);
    ReopenWithColumnFamilies({"default", "pikachu"}, puremem_options);
    ASSERT_OK(Put(1, "foo", "v3"));
    ReopenWithColumnFamilies({"default", "pikachu"}, puremem_options);
    ASSERT_EQ("v3", Get(1, "foo"));
  } while (ChangeWalOptions());
}

TEST_F(DBPurememWALTest, GetSortedWalFiles) {
  do {
    auto puremem_options = GetPurememOptions();
    CreateAndReopenWithCF({"pikachu"}, puremem_options);
    VectorLogPtr log_files;
    ASSERT_OK(dbfull()->GetSortedWalFiles(log_files));
    ASSERT_EQ(0, log_files.size());

    ASSERT_OK(Put(1, "foo", "v1"));
    ASSERT_OK(dbfull()->GetSortedWalFiles(log_files));
    ASSERT_EQ(1, log_files.size());
  } while (ChangeWalOptions());
}

TEST_F(DBPurememWALTest, RecoveryWithLogDataForSomeCFs) {
  // Test for regression of WAL cleanup missing files that don't contain data
  // for every column family.
  do {
    auto puremem_options = GetPurememOptions();
    CreateAndReopenWithCF({"pikachu"}, puremem_options);
    ASSERT_OK(Put(1, "foo", "v1"));
    ReopenWithColumnFamilies({"default", "pikachu"}, puremem_options);
    ASSERT_OK(Put(1, "foo", "v2"));
    uint64_t earliest_log_nums[2];
    for (int i = 0; i < 2; ++i) {
      if (i > 0) {
        ReopenWithColumnFamilies({"default", "pikachu"}, puremem_options);
      }
      VectorLogPtr log_files;
      ASSERT_OK(dbfull()->GetSortedWalFiles(log_files));
      if (log_files.size() > 0) {
        earliest_log_nums[i] = log_files[i]->LogNumber();
      } else {
        earliest_log_nums[i] = port::kMaxUint64;
      }
    }
    // Check at least the first WAL was cleaned up during the recovery.
    ASSERT_LT(earliest_log_nums[0], earliest_log_nums[1]);
  } while (ChangeWalOptions());
}


TEST_F(DBPurememWALTest, RecoverWithLargeLog) {
  do {
    auto puremem_options = GetPurememOptions();
    {
      CreateAndReopenWithCF({"pikachu"}, puremem_options);
      ASSERT_OK(Put(1, "big1", std::string(200000, '1')));
      ASSERT_OK(Put(1, "big2", std::string(200000, '2')));
      ASSERT_OK(Put(1, "small3", std::string(10, '3')));
      ASSERT_OK(Put(1, "small4", std::string(10, '4')));
    }

    // Make sure that if we re-open with a small write buffer size that
    // we flush table files in the middle of a large log file.

    ReopenWithColumnFamilies({"default", "pikachu"}, puremem_options);
    ASSERT_EQ(std::string(200000, '1'), Get(1, "big1"));
    ASSERT_EQ(std::string(200000, '2'), Get(1, "big2"));
    ASSERT_EQ(std::string(10, '3'), Get(1, "small3"));
    ASSERT_EQ(std::string(10, '4'), Get(1, "small4"));
  } while (ChangeWalOptions());
}

TEST_F(DBPurememWALTest, PartOfWritesWithWALDisabled) {
  auto puremem_options = GetPurememOptions();
  WriteOptions wal_on, wal_off;
  wal_on.disableWAL = false;
  wal_off.disableWAL = true;
  CreateAndReopenWithCF({"dummy"}, puremem_options);
  ASSERT_OK(Put(1, "dummy", "d1", wal_on));  // seq id 1
  ASSERT_OK(Put(1, "dummy", "d2", wal_off));
  ASSERT_OK(Put(1, "dummy", "d3", wal_on));
  ASSERT_OK(Put(0, "key", "v4", wal_on));  // seq id 4
  ASSERT_OK(Put(0, "key", "v5", wal_on));  // seq id 5
  ASSERT_EQ("d3", Get(1, "dummy"));
  ASSERT_EQ("v5", Get(0, "key"));

  Close();
  ReopenWithColumnFamilies({"default", "dummy"}, puremem_options);
  // Prior to the fix, we may incorrectly recover "v5" with sequence id = 3.
  ASSERT_EQ("v5", Get(0, "key"));
  ASSERT_EQ("d3", Get(1, "dummy"));
  // Destroy DB before destruct fault_env.
  Destroy(puremem_options);
}


TEST_F(DBPurememWALTest, WalTermTest) {
  auto puremem_options = GetPurememOptions();
  puremem_options.env = env_;
  CreateAndReopenWithCF({"pikachu"}, puremem_options);

  ASSERT_OK(Put(1, "foo", "bar"));

  WriteOptions wo;
  wo.sync = true;
  wo.disableWAL = false;

  WriteBatch batch;
  batch.Put("foo", "bar");
  batch.MarkWalTerminationPoint();
  batch.Put("foo2", "bar2");

  ASSERT_OK(dbfull()->Write(wo, &batch));

  // make sure we can re-open it.
  ASSERT_OK(TryReopenWithColumnFamilies({"default", "pikachu"}, puremem_options));
  ASSERT_EQ("bar", Get(1, "foo"));
  ASSERT_EQ("NOT_FOUND", Get(1, "foo2"));
}

std::mutex mu;
void WriteDataTest(DBPurememWALTest *threadarg)
{
  // Jacob: 对传入的参数进行强制类型转换，由无类型指针变为整形数指针，然后再读取
  static int threads_id =0;
  mu.lock();
  threads_id++;
  int thread_id =threads_id;
  mu.unlock();
  DBPurememWALTest *w_data;
  w_data = threadarg;
  for (int k = 0; k < 100; ++k) {
    int key = thread_id * 1000 + k * 10;
    ASSERT_OK(w_data->Put(ToString(key), std::string(2, 'v')));
    ASSERT_OK(w_data->Put(ToString(key+1), "v1"));
    ASSERT_OK(w_data->Put(ToString(key+2), "v2"));
    ASSERT_OK(w_data->Put(ToString(key+3), "v3"));
    ASSERT_OK(w_data->Put(ToString(key+4), "v4"));
  };
}

TEST_F(DBPurememWALTest, MultiWritePurememTest){
  auto puremem_options = GetPurememOptions();
  Reopen(puremem_options);
  std::thread threads[20];
  for (int i = 0; i < 20; ++i)
    threads[i] = std::thread(WriteDataTest, this);
  for (auto &thread : threads)
    thread.join();

  ASSERT_EQ("vv",Get(ToString(2850)));
  ASSERT_EQ("v2",Get(ToString(12602)));
}
}

int main(int argc, char** argv) {
  rocksdb::port::InstallStackTraceHandler();
  ::testing::InitGoogleTest(&argc, argv);
  return RUN_ALL_TESTS();
}